Alarmer ved latenstid, omkostninger og kvalitetsforringelse
Definér alarmgrænser for p99-latenstid, omkostninger pr. forespørgsel og automatiserede kvalitetsscorer, og send alarmer til Slack eller PagerDuty, når Deres LLM-pipeline forringes.
Alarmer ved latenstid, omkostninger og kvalitetsforringelse er en gratis AI Engineering Academy-lektion på CoddyKit. Dette er lektion 4 af 4. Du kan læse hele lektionen gratis nedenfor — og derefter øve dig praktisk i browseren med en indbygget kodeeditor og en AI-vejleder, der er tilgængelig døgnet rundt. Den er en del af læringsforløbet i AI Engineering Academy, og dine fremskridt synkroniseres på tværs af nettet og CoddyKit-appen. AI Engineering Academy-kurset indeholder 4 lektioner i alt.
Hvorfor alarmer er vigtige for LLM-systemer
LLM-applikationer svigter på måder, der er subtile og gradvise. En promptændring kan øge den gennemsnitlige latenstid med 30 %, en forbedring af hentningen kan reducere svarkvaliteten en smule, eller en stigning i brugen kan presse de daglige omkostninger op til fem gange over budgettet. Uden proaktive alarmer opdager du først disse problemer, når brugerne klager, eller den månedlige regning kommer. Alarmer omdanner reaktiv brandslukning til proaktiv drift.
De tre alarmkategorier
Alarmer i LLM-applikationer falder i tre kategorier. Latenstidsalarmer udløses, når svartiden overskrider en grænse for brugeroplevelsen (f.eks. p99 > 10 sekunder). Omkostningsalarmer udløses, når omkostningen pr. forespørgsel eller det daglige forbrug overskrider budgetgrænser, så uventet høje regninger forebygges. Kvalitetsalarmer udløses, når automatiske kvalitetsmålinger (scorer fra LLM som bedømmer, brugertilfredshed og troværdighed) falder under et acceptabelt minimum. Hver kategori kræver forskellig instrumentering og forskellige alarmkanaler.
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()Indsamling af latenstidsmålinger
Hvis du vil udløse alarmer på baggrund af latenstid, skal du først indsamle den konsekvent. Mål tre latenstidskomponenter separat: tid til første token (TTFT, afgørende for brugeroplevelsen ved streaming), samlet svartid og latenstid pr. trin for hentning og generering. Gem disse som tidsseriedata (ét datapunkt pr. forespørgsel), så du kan beregne percentiler og rullende gennemsnit for de seneste N minutter eller timer.
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()Sporing og alarmering på omkostninger
Omkostningsovervågning kræver sporing på flere detaljeringsniveauer: omkostning pr. forespørgsel (for at opdage usædvanligt dyre forespørgsler), time- og dagstotaler (for at opdage bratte stigninger i brugen) samt omkostning pr. funktion (for at identificere den dyreste del af applikationen). Udløs straks en alarm ved uregelmæssigheder i omkostningen pr. forespørgsel, og brug planlagte kontroller til daglige budgetgrænser.
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()Alarmering på kvalitetsscorer
Kvalitetsalarmering er den mest komplekse kategori, fordi kvalitet ikke kan måles ud fra infrastrukturmålinger alene — det kræver semantisk evaluering. Den mest skalerbare tilgang er automatisk evaluering af stikprøver: Kør en evaluator, hvor en LLM fungerer som bedømmer, asynkront på et tilfældigt udsnit af produktionsforespørgsler (5-10 %), gem scorerne, og beregn et rullende gennemsnit. Udløs en alarm, når det rullende gennemsnit falder under dit kvalitetsminimum.
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 NoneRullende gennemsnit til registrering af tendenser
Ét dårligt svar er ikke tegn på et systemisk problem. Brug rullende gennemsnit over et tidsvindue (f.eks. den seneste time eller de seneste 100 forespørgsler) til at registrere tendenser. Et rullende gennemsnit, der krydser en grænse og bliver der i mere end 10 minutter, tyder på et reelt problem, mens en kortvarig stigning kan være støj. Eksponentielt vægtede glidende gennemsnit (EWMA) reagerer hurtigere på nylige ændringer end simple rullende gennemsnit.
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()Alarmdirigering: Slack og PagerDuty
Alarmer er kun nyttige, hvis de når den rigtige person via den rigtige kanal. Dirigér alarmer efter alvorlighed: Alarmer om forringet kvalitet og overskridelse af omkostningsbudgettet sendes til en Slack-kanal, som teamet overvåger i arbejdstiden. Latenstidsalarmer over kritiske grænser (f.eks. p99 > 30 sekunder) og pludselige omkostningsstigninger (f.eks. 10 gange det normale på 5 minutter) sendes til PagerDuty, så den vagthavende straks eskaleres.
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)Løkken til alarmevaluering
Alarmlogik bør køre i en planlagt løkke, ikke direkte under behandlingen af forespørgsler. Asynkrone alarmkontroller forhindrer, at de tilføjer latenstid til produktionsforespørgsler. En baggrundstråd eller et planlagt job, der kører hvert 60. sekund, er tilstrækkeligt til de fleste alarmbehov. Indsaml målinger i forespørgselsforløbet, og evaluer grænserne i baggrundsløkken.
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()Alarmtræthed og justering af grænser
Alarmtræthed opstår, når alarmer udløses så ofte, at teamet begynder at ignorere dem. Forebyg det ved at starte med konservative (høje) grænser og stramme dem, efterhånden som du forstår dit normale niveau, kræve at alarmer varer ved gennem flere evalueringscyklusser, før de udløses, samle relaterede alarmer i én meddelelse og regelmæssigt gennemgå og fjerne alarmer, der aldrig fører til meningsfuld handling.
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)Overvågning af omkostningsafvigelser med statistiske metoder
Simple grænsealarmer overser nuancerede omkostningsafvigelser som en gradvis omkostningsstigning på 50 % over to dage. Brug statistisk registrering af afvigelser: Beregn det rullende gennemsnit og standardafvigelsen for dine timeomkostninger, og udløs en alarm, når den aktuelle times omkostning ligger mere end N standardafvigelser over gennemsnittet (Z-score-alarmering). Det tilpasser sig automatisk til naturlige brugsmønstre som trafik på hverdage og i weekender.
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 NoneOpbygning af et driftsdashboard
Alarmer er det reaktive lag; et live-driftsdashboard er det proaktive lag. Byg et enkelt dashboard, der viser målinger i realtid: aktuel p50-/p99-latenstid, forespørgsler pr. minut, aktuel kvalitetsscore (rullende EWMA), daglige omkostninger til dato sammenholdt med budgettet samt de fem dyreste forespørgsler i den seneste time. Det giver teamet et hurtigt overblik over systemets tilstand uden at vente på, at en alarm udløses.
Hurtigt tjek
Test din forståelse af alarmering på målinger for LLM-applikationer fra denne lektion.
Opsummering af lektionen
I denne lektion lærte du: tre alarmkategorier dækker LLM-systemer — svartid, omkostninger og kvalitet — og hver kategori kræver forskellig instrumentering; glidende gennemsnit og EWMA opdager gradvis kvalitetsforringelse bedre end tærskelkontrol af individuelle forespørgsler; og alarmdebouncing forebygger alarmtræthed ved at kræve, at betingelserne varer ved gennem flere evalueringscyklusser, før alarmen udløses. I næste lektion dykker vi ned i promptinjektionsangreb og AI-sikkerhed.
Lær Python med en AI-underviser — gratis
Skriv og kør rigtig kode i din browser, få øjeblikkelig hjælp fra en AI-underviser døgnet rundt, og fortsæt, hvor du slap, på web eller i appen.
- Kurser
- 30
- Lektioner
- 120
Ofte stillede spørgsmål
Er lektionen “Alarmer ved latenstid, omkostninger og kvalitetsforringelse” gratis?
Ja — hele teksten til “Alarmer ved latenstid, omkostninger og kvalitetsforringelse” kan læses gratis her på nettet. Hvis du vil øve dig interaktivt med en indbygget kodeeditor og en AI-vejleder døgnet rundt og få adgang til resten af AI Engineering Academy-kurset, skal du opgradere til CoddyKit PRO. AI Engineering Academy-kurset indeholder 4 lektioner i alt.
Hvad lærer jeg i “Alarmer ved latenstid, omkostninger og kvalitetsforringelse”?
Definér alarmgrænser for p99-latenstid, omkostninger pr. forespørgsel og automatiserede kvalitetsscorer, og send alarmer til Slack eller PagerDuty, når Deres LLM-pipeline forringes. Du øver dig i AI Engineering Academy med praktisk kode, som du kører direkte i browseren, og en AI-vejleder døgnet rundt besvarer dine spørgsmål, mens du arbejder dig gennem lektionen.
Skal jeg have erfaring for at begynde på AI Engineering Academy?
Der kræves ingen tidligere erfaring. AI Engineering Academy på CoddyKit er tilrettelagt for både begyndere og øvede, så du kan starte her eller fra begyndelsen og lære i dit eget tempo. Dette er lektion 4 af 4.
Hvor lang tid tager lektionen “Alarmer ved latenstid, omkostninger og kvalitetsforringelse”?
De fleste CoddyKit-lektioner tager cirka 5–10 minutter. Hver lektion er kort og interaktiv, så du gør løbende fremskridt og kan fortsætte, hvor du slap – på både web og app.
Kan jeg skrive og køre kode i denne AI Engineering Academy-lektion?
Ja. Alle AI Engineering Academy-lektioner har en indbygget kodeeditor, så du kan skrive og køre rigtig kode direkte i din browser og få øjeblikkelig feedback fra AI – uden lokal opsætning.
Alle lektioner i dette kursus
- Hvorfor LLM-apps er svære at fejlfinde
- Tracing med LangSmith
- Langfuse til modeluafhængig observability
- Alarmer ved latenstid, omkostninger og kvalitetsforringelse