Promptteknik til AI · Lektion

Overvågning og alarmering for promptpipelines

Dashboards, anomalidetektion og on-call-alarmer for prompts i produktion.

Lektion 4 af 413 trin

Overvågning og alarmering for promptpipelines er en gratis Promptteknik til AI-lektion på CoddyKit. Dette er lektion 4 af 4. Du kan læse alle 3 lektioner i dette læringsspor gratis i deres fulde længde — derefter låser CoddyKit PRO alle lektioner op samt praktiske øvelser med en indbygget kodeeditor og en AI-underviser døgnet rundt. Den er en del af læringsforløbet i Promptteknik til AI, og dine fremskridt synkroniseres på tværs af nettet og CoddyKit-appen. Promptteknik til AI-kurset indeholder 4 lektioner i alt.

Promptforløb i produktion kræver overvågning

Et promptforløb i produktion er infrastruktur — det har brug for kontrolpaneler, alarmer og driftsvejledninger ligesom enhver anden tjeneste. Uden overvågning bliver omkostningsstigninger, kvalitetsforringelser og kraftigt stigende latenstid ikke opdaget, før brugerne klager, eller regningerne kommer.

Centrale målinger: latenstidspercentiler

Spor latenstid ved P50, P95 og P99. Gennemsnittet skjuler yderpunkternes adfærd — en P99 på 30 sekunder betyder, at 1 % af brugerne venter et halvt minut, selv om P50 er 2 sekunder. LLM-latenstid varierer uundgåeligt, fordi den skalerer med uddataens længde.

import time
import statistics
from collections import deque

class LatencyTracker:
    def __init__(self, window_size=1000):
        self.samples = deque(maxlen=window_size)

    def record(self, latency_ms):
        self.samples.append(latency_ms)

    def percentile(self, p):
        if not self.samples:
            return None
        sorted_samples = sorted(self.samples)
        idx = int(len(sorted_samples) * p / 100)
        return sorted_samples[min(idx, len(sorted_samples) - 1)]

    def report(self):
        if not self.samples:
            return {}
        return {
            'count': len(self.samples),
            'p50_ms': self.percentile(50),
            'p95_ms': self.percentile(95),
            'p99_ms': self.percentile(99),
            'max_ms': max(self.samples)
        }

tracker = LatencyTracker()
for ms in [1200, 1100, 1300, 1150, 8500, 1200, 1250, 15000, 1100, 1300]:
    tracker.record(ms)
print(tracker.report())

Måling af daglige omkostninger i kontrolpanelet

Spor de daglige API-omkostninger som en central måling i kontrolpanelet. Beregn dem ud fra tokenforbruget i realtid, og sammenlign med det rullende ugentlige gennemsnit for at opdage omkostningsstigninger tidligt.

from datetime import datetime, timedelta
from collections import defaultdict

class CostTracker:
    def __init__(self):
        self.daily_costs = defaultdict(float)  # date: total_cost_usd

    def record_call(self, model, input_tokens, output_tokens):
        pricing = {
            'gpt-4o-mini': (0.15, 0.60),
            'gpt-4o': (2.50, 10.00),
            'claude-opus-4-5': (15.00, 75.00),  # per 1M tokens
            'claude-haiku-4-5': (0.25, 1.25)
        }
        if model not in pricing:
            return
        input_price, output_price = pricing[model]
        cost = (input_tokens / 1_000_000 * input_price +
                output_tokens / 1_000_000 * output_price)
        today = datetime.utcnow().date().isoformat()
        self.daily_costs[today] += cost

    def today_cost(self):
        today = datetime.utcnow().date().isoformat()
        return round(self.daily_costs[today], 4)

    def weekly_avg_daily_cost(self):
        dates = sorted(self.daily_costs.keys())[-7:]
        if not dates:
            return 0
        return round(sum(self.daily_costs[d] for d in dates) / len(dates), 4)

cost_tracker = CostTracker()
cost_tracker.record_call('gpt-4o-mini', 800, 200)
print('Today cost:', cost_tracker.today_cost())

Overvågning af fejlrate

Spor fejlratens procentdel af det samlede antal forespørgsler. Fejl omfatter API-fejl, timeouts, forkert uddata, der ikke kan fortolkes, og modelafvisninger. Adskil fejltyperne, så alarmerne kan udløse konkrete handlinger.

from collections import Counter

class ErrorRateTracker:
    ERROR_TYPES = [
        'api_error', 'timeout', 'rate_limit',
        'parse_failure', 'model_refusal', 'context_length_exceeded'
    ]

    def __init__(self, window_size=1000):
        self.total = 0
        self.errors = Counter()
        self.recent = deque(maxlen=window_size)  # True=error, False=success

    def record(self, success, error_type=None):
        self.total += 1
        self.recent.append(not success)
        if not success and error_type:
            self.errors[error_type] += 1

    def error_rate(self):
        if not self.recent:
            return 0.0
        return sum(self.recent) / len(self.recent)

    def report(self):
        return {
            'error_rate': round(self.error_rate(), 4),
            'total_requests': self.total,
            'error_breakdown': dict(self.errors.most_common())
        }

err_tracker = ErrorRateTracker()
for i in range(100):
    if i % 20 == 0:
        err_tracker.record(False, 'timeout')
    else:
        err_tracker.record(True)
print(err_tracker.report())

Sporing af udviklingen i kvalitetsscoren

Spor kvalitetsscoren som en rullende tidsserie, så udviklingen bliver synlig. Et gradvist kvalitetsfald over flere dage er sværere at opdage end et pludseligt fald, men kan skade brugernes tillid lige så meget.

from datetime import datetime
import statistics

class QualityTrendTracker:
    def __init__(self, window_minutes=60):
        self.window_seconds = window_minutes * 60
        self.samples = []  # (timestamp, score)

    def record(self, score):
        now = time.time()
        self.samples.append((now, score))
        # Purge old samples outside window
        cutoff = now - self.window_seconds
        self.samples = [(t, s) for t, s in self.samples if t >= cutoff]

    def rolling_avg(self):
        if not self.samples:
            return None
        return round(statistics.mean(s for _, s in self.samples), 3)

    def trend(self):
        if len(self.samples) < 10:
            return 'insufficient_data'
        mid = len(self.samples) // 2
        first_half_avg = statistics.mean(s for _, s in self.samples[:mid])
        second_half_avg = statistics.mean(s for _, s in self.samples[mid:])
        delta = second_half_avg - first_half_avg
        if delta > 0.1:
            return 'improving'
        elif delta < -0.1:
            return 'declining'
        return 'stable'

qt = QualityTrendTracker(window_minutes=60)
for score in [4.2, 4.1, 4.0, 3.9, 3.8, 3.7, 3.6, 3.5, 3.4, 3.3]:
    qt.record(score)
print('Rolling avg:', qt.rolling_avg(), '| Trend:', qt.trend())

Udformning af kontrolpaneler

Et velformet overvågningskontrolpanel grupperer målinger i logiske sektioner. Definér de fire centrale paneler, som alle kontrolpaneler for promptforløb har brug for.

DASHBOARD_PANELS = {
    'Panel 1: Availability': [
        'Error rate (%) — last 1h, 24h, 7d',
        'Error type breakdown (timeout vs API vs parse)',
        'P99 latency (alert if > 10s)',
        'Success rate by prompt_id and version'
    ],
    'Panel 2: Performance': [
        'P50 / P95 / P99 latency (time series)',
        'Latency by model and prompt version',
        'Time to first token (streaming)',
        'Latency heatmap by hour of day'
    ],
    'Panel 3: Cost': [
        'Daily cost USD (actual vs budget)',
        'Cost per request by model',
        'Cost trend (7-day rolling)',
        'Top 10 most expensive prompt_ids'
    ],
    'Panel 4: Quality': [
        'Average quality score (rolling 1h)',
        'Quality trend by prompt version',
        'User satisfaction (thumbs, retry rate)',
        'Low-quality alert rate'
    ]
}

for panel, metrics in DASHBOARD_PANELS.items():
    print(f'\n{panel}:')
    for m in metrics:
        print(f'  - {m}')

Implementering af alarmregler

Alarmer udløses, når målinger overskrider tærskler. Implementér alarmer som enkle statuskontroller, der kører efter en tidsplan og sender notifikationer til PagerDuty, Slack eller e-mail.

ALERT_RULES = [
    {
        'name': 'HighErrorRate',
        'condition': lambda m: m['error_rate'] > 0.05,
        'severity': 'CRITICAL',
        'message': 'Error rate {error_rate:.1%} exceeds 5% threshold',
        'for_minutes': 5
    },
    {
        'name': 'HighLatencyP95',
        'condition': lambda m: m.get('p95_latency_ms', 0) > 10000,
        'severity': 'WARNING',
        'message': 'P95 latency {p95_latency_ms}ms exceeds 10s threshold',
        'for_minutes': 3
    },
    {
        'name': 'CostSpike',
        'condition': lambda m: m.get('today_cost', 0) > m.get('weekly_avg', 1) * 2,
        'severity': 'WARNING',
        'message': 'Daily cost ${today_cost:.2f} is 2x weekly average',
        'for_minutes': 60
    },
    {
        'name': 'QualityRegression',
        'condition': lambda m: m.get('quality_avg', 5) < 3.5,
        'severity': 'CRITICAL',
        'message': 'Quality score {quality_avg:.2f} below 3.5 threshold',
        'for_minutes': 15
    }
]

def check_alerts(metrics):
    fired = []
    for rule in ALERT_RULES:
        if rule['condition'](metrics):
            msg = rule['message'].format(**metrics)
            fired.append({'name': rule['name'], 'severity': rule['severity'],
                           'message': msg})
    return fired

Udsendelse af notifikationer

Alarmnotifikationer skal dirigeres efter alvorlighed: CRITICAL-alarmer tilkalder straks den vagthavende; WARNING-alarmer sendes til Slack; INFO-alarmer skrives til en logfil. Brug eskaleringstimere for kritiske alarmer, der ikke bliver anerkendt.

import requests

SLACK_WEBHOOK = 'https://hooks.slack.com/services/YOUR/WEBHOOK/URL'
PAGERDUTY_API_KEY = 'YOUR_PD_KEY'

def send_slack_alert(message, severity='WARNING'):
    emoji = ':rotating_light:' if severity == 'CRITICAL' else ':warning:'
    payload = {'text': f'{emoji} *{severity}*: {message}'}
    try:
        requests.post(SLACK_WEBHOOK, json=payload, timeout=5)
        print(f'Slack alert sent: {message[:60]}')
    except Exception as e:
        print(f'Slack notification failed: {e}')

def send_pagerduty_alert(summary, severity='critical'):
    payload = {
        'routing_key': PAGERDUTY_API_KEY,
        'event_action': 'trigger',
        'payload': {
            'summary': summary,
            'severity': severity,
            'source': 'prompt-pipeline-monitor'
        }
    }
    try:
        response = requests.post(
            'https://events.pagerduty.com/v2/enqueue',
            json=payload, timeout=10
        )
        print(f'PagerDuty alert: {response.status_code}')
    except Exception as e:
        print(f'PagerDuty notification failed: {e}')

def dispatch_alert(alert):
    if alert['severity'] == 'CRITICAL':
        send_pagerduty_alert(alert['message'])
        send_slack_alert(alert['message'], 'CRITICAL')
    else:
        send_slack_alert(alert['message'], 'WARNING')

Struktur for driftsvejledninger

Alle alarmer bør have en tilhørende driftsvejledning, der fortæller den vagthavende udvikler præcis, hvad vedkommende skal gøre. Velskrevet dokumentation reducerer MTTR (gennemsnitlig løsningstid) fra timer til minutter.

# Runbook template for HighErrorRate alert
HIGH_ERROR_RATE_RUNBOOK = '''
## Alert: HighErrorRate
### Trigger: Error rate > 5% for > 5 minutes
### Severity: CRITICAL

## Immediate Actions (< 5 minutes)
1. Check error type breakdown in dashboard: Panel 1 > Error type breakdown
   - timeout errors -> see Timeout Runbook
   - api_error -> check LLM provider status page
   - parse_failure -> check if model output format changed

2. Check if this is related to a recent deployment:
   python manage.py prompt list-recent-activations --last-hours 2

3. If error rate > 20%, trigger emergency rollback:
   python manage.py prompt activate --prompt-id <id> --version <last-stable>

## Investigation (< 30 minutes)
4. Sample failed requests from log:
   grep error_rate /var/log/prompt-pipeline.log | tail -100

5. Check model provider status:
   - OpenAI: https://status.openai.com
   - Anthropic: https://status.anthropic.com

## Resolution
6. If provider outage: activate fallback model routing
7. If prompt change: rollback to previous version
8. If code change: rollback deployment
9. Document in post-mortem after resolution
'''

print(HIGH_ERROR_RATE_RUNBOOK[:400], '...')

Struktureret logning for promptforløb

Strukturerede logge (JSON pr. linje) gør det muligt at filtrere og aggregere effektivt i logstyringssystemer som Datadog, Splunk eller CloudWatch. Alle LLM-kald bør producere én struktureret logpost.

import json
import time
from datetime import datetime

def log_llm_call(request_id, prompt_id, version, model, messages,
                  response_text, latency_ms, input_tokens,
                  output_tokens, error=None):
    log_entry = {
        'ts': datetime.utcnow().isoformat() + 'Z',
        'level': 'ERROR' if error else 'INFO',
        'service': 'prompt-pipeline',
        'request_id': request_id,
        'prompt_id': prompt_id,
        'version': version,
        'model': model,
        'latency_ms': round(latency_ms),
        'input_tokens': input_tokens,
        'output_tokens': output_tokens,
        'error': str(error) if error else None,
        'response_preview': response_text[:100] if response_text else None
    }
    print(json.dumps(log_entry))
    # In production: ship to log aggregator
    # logger.info(json.dumps(log_entry))

# Example log output:
# {"ts":"2024-08-15T10:00:01Z","level":"INFO",
#  "prompt_id":"summarize-article","version":"1.2.0",
#  "model":"gpt-4o-mini","latency_ms":1234,
#  "input_tokens":800,"output_tokens":150,...}
log_llm_call('req-001', 'summarize-article', '1.2.0', 'gpt-4o-mini',
             [], 'Summary text...', 1234, 800, 150)

Arkitektur for overvågningssystemet

Saml alle overvågningskomponenter i et sammenhængende system, der kører sammen med promptforløbet. En enkel løkke med statuskontroller håndterer evaluering og udsendelse af alarmer.

import time

class PromptPipelineMonitor:
    def __init__(self):
        self.latency = LatencyTracker()
        self.errors = ErrorRateTracker()
        self.quality = QualityTrendTracker()
        self.cost = CostTracker()

    def record(self, model, latency_ms, input_tokens, output_tokens,
                quality_score=None, error=None, error_type=None):
        self.latency.record(latency_ms)
        self.errors.record(error is None, error_type)
        self.cost.record_call(model, input_tokens, output_tokens)
        if quality_score:
            self.quality.record(quality_score)

    def current_metrics(self):
        lat = self.latency.report()
        err = self.errors.report()
        return {
            **lat, **err,
            'quality_avg': self.quality.rolling_avg() or 5.0,
            'quality_trend': self.quality.trend(),
            'today_cost': self.cost.today_cost(),
            'weekly_avg': self.cost.weekly_avg_daily_cost()
        }

    def run_alert_check(self):
        metrics = self.current_metrics()
        alerts = check_alerts(metrics)
        for alert in alerts:
            dispatch_alert(alert)
        return alerts

monitor = PromptPipelineMonitor()
print('Monitor initialized. Call monitor.record() on each LLM call.')

Hurtigt tjek

Dit promptforløbs latenstid ved P50 er 1,5 sekunder, men P99 er 28 sekunder. Hvad fortæller det dig om adfærden i produktionen?

Besvar den i den interaktive lektion

Opsummering af overvågning og alarmer

Overvågning af promptforløb i produktion kræver fem målekategorier med tilhørende alarmregler:

Latenstid: spor P50/P95/P99 — alarm når P95 > 10 sek.
Omkostninger: daglige omkostninger sammenlignet med det ugentlige gennemsnit — alarm ved en fordobling af omkostningerne
Fejlrate: samlet procentdel og fordeling på fejltyper — alarm når > 5 %
Kvalitetsscore: rullende gennemsnitlig udvikling — alarm når den falder under 3,5/5
Alarmer: CRITICAL → PagerDuty + Slack; WARNING → kun Slack
Driftsvejledninger: trin-for-trin-vejledninger til løsning af alle alarmtyper
Kontrolpanel: fire paneler, der dækker tilgængelighed, ydeevne, omkostninger og kvalitet

Gratis at komme i gang

Lær Promptteknik til AI 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: 53
Lektioner: 199

Start gratis Se hele kurset

Ofte stillede spørgsmål

Er lektionen “Overvågning og alarmering for promptpipelines” gratis?

Ja — alle 3 lektioner i læringssporet Promptteknik til AI, inklusive “Overvågning og alarmering for promptpipelines”, kan læses gratis i deres fulde længde her på webstedet. Derefter låser CoddyKit PRO alle lektioner op samt interaktive øvelser med en indbygget kodeeditor og en AI-underviser døgnet rundt. Promptteknik til AI-kurset indeholder 4 lektioner i alt.

Hvad lærer jeg i “Overvågning og alarmering for promptpipelines”?

Dashboards, anomalidetektion og on-call-alarmer for prompts i produktion. Du øver dig i Promptteknik til AI 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å Promptteknik til AI?

Der kræves ingen tidligere erfaring. Promptteknik til AI 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 “Overvågning og alarmering for promptpipelines”?

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 Promptteknik til AI-lektion?

Ja. Alle Promptteknik til AI-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

Cachingstrategier for prompts
Batchbehandling og asynkron eksekvering
Load balancing på tværs af modeller
Overvågning og alarmering for promptpipelines

← Tilbage til Promptteknik til AI